package com.tmsb.sink;

import com.tmsb.pojo.ESScore;
import com.tmsb.utils.*;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.JedisCluster;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

/**
 * Description: 用es的score表的获得每个省的日贡献值和日贡献值两个实时指标的sink
 * Version:1.0.0
 *
 * @author xiekb
 * @date 2020/7/3
 */
public class ScoreSink extends RichSinkFunction<ESScore> implements CheckpointedFunction {
    private ListState<Long> allAmountValueState;
    private ListState<Long> dateState;
    private BroadcastState<String, Long> cityAmountMapState;
    private Long allAmountValue;
    private Long dateValue;
    private Map<String,Long> cityAmountMap;
    private JedisCluster jedisCluster;
    private SimpleDateFormat sdf;


    @Override
    public void invoke(ESScore value, Context context) throws Exception {
        //每天0时清空状态
        long eventTime = value.getLast();
        if(eventTime>= dateValue){
            allAmountValue = 0L;
            cityAmountMap.clear();
            dateValue = Util.getTomorrowZeroPointTimestamps();
        }else if(eventTime< Util.getZeroPointTimestamps()){
            return;
        }


        //每日总贡献值
        int amount = value.getAmount();
        allAmountValue = allAmountValue + amount;
        //每日城市贡献值
        String cityId = value.getCity_id()+"";
        String provinceKey = "";
        long scoreCityAmount = 0L;
        if (cityId.length()== Const.CITYID_LENGTH) {
            provinceKey = ProvinceCodeTrans.getProvinceKey(cityId);
            scoreCityAmount = cityAmountMap.getOrDefault(provinceKey, 0L) + amount;
            cityAmountMap.put(provinceKey, scoreCityAmount);
        }

        //每三毫秒向redis发送一次数据
        if(Math.floorMod(eventTime,3)==0){
            String format = sdf.format(new Date(eventTime));
            jedisCluster.hset(ConfUtil.getScoreAmountRedisKey(),format,allAmountValue+"");
            if (cityId.length()== Const.CITYID_LENGTH) {
                jedisCluster.zadd(ConfUtil.getScoreCityAmountRedisKey() + format, scoreCityAmount, provinceKey);
            }
        }

    }

    @Override
    public void close() throws Exception {
        super.close();
        jedisCluster.close();
    }


    //当请求 checkpoint 快照时，将调用此方法
    @Override
    public void snapshotState(FunctionSnapshotContext functionSnapshotContext) throws Exception {
        allAmountValueState.clear();
        dateState.clear();
        cityAmountMapState.clear();
        allAmountValueState.add(allAmountValue);
        dateState.add(dateValue);
        cityAmountMapState.putAll(cityAmountMap);
    }

    //在分布式执行期间创建并行功能实例时，将调用此方法。
    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        jedisCluster = RedisCluster.getInstance(ConfUtil.getRedisCluster()).getRedisClusterList().get(0);
        sdf = new SimpleDateFormat("yyyyMMdd");
        dateValue = 0L;
        allAmountValue = 0L;
        cityAmountMap = new HashMap<String,Long>();
        ListStateDescriptor<Long> allstateDes =
                new ListStateDescriptor<>("allstate", TypeInformation.of(new TypeHint<Long>() {}));
        ListStateDescriptor<Long> dateStateDes =
                new ListStateDescriptor<>("date", TypeInformation.of(new TypeHint<Long>() {}));
        MapStateDescriptor<String, Long> CityStateDescriptor = new MapStateDescriptor<>(
                "CityStateDescriptor",
                BasicTypeInfo.STRING_TYPE_INFO,
                TypeInformation.of(new TypeHint<Long>() {}));
        allAmountValueState = context.getOperatorStateStore().getListState(allstateDes);
        dateState = context.getOperatorStateStore().getListState(dateStateDes);
        cityAmountMapState = context.getOperatorStateStore().getBroadcastState(CityStateDescriptor);
        if (context.isRestored()) {
            allAmountValue = allAmountValueState.get().iterator().next();
            dateValue = dateState.get().iterator().next();
            Iterator<Map.Entry<String, Long>> cityAmountIterator = cityAmountMapState.iterator();
            while(cityAmountIterator.hasNext()){
                Map.Entry<String, Long> next = cityAmountIterator.next();
                cityAmountMap.put(next.getKey(),next.getValue());
            }
        }
    }
}
